package com.iussoft.demo.jms;


import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;

import com.iussoft.demo.util.ConfigUtil;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;



/**
 * @description: 接收消息客户端代码
 * @author: Hou Dayu
 * @date: Created in 2020/10/27
 */
public class QueueReceiver implements MessageListener {

    private ConnectionFactory connFactory;
    private Connection conn;
    private Session session;
    private boolean stop=false;

    public Session getSession() {
        return session;
    }

    public void execute()throws Exception{
        //连接工厂
        // 设置用户名和密码，这个用户名和密码在conf目录下的credentials.properties文件中
        connFactory=new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                ConfigUtil.getConfigByKey("jms.connector.uri"));
        //连接到JMS提供者
        conn=connFactory.createConnection();
        conn.start();

        //事务性会话，自动确认消息
        // 第一个参数是否使用事务:当消息发送者向消息提供者（即消息代理）发送消息时，消息发送者等待消息代理的确认，没有回应则抛出异常，消息发送程序负责处理这个错误。
        // 第二个参数消息的确认模式：
        // AUTO_ACKNOWLEDGE ： 指定消息提供者在每次收到消息时自动发送确认。消息只向目标发送一次，但传输过程中可能因为错误而丢失消息。
        // CLIENT_ACKNOWLEDGE ： 由消息接收者确认收到消息，通过调用消息的acknowledge()方法（会通知消息提供者收到了消息）
        // DUPS_OK_ACKNOWLEDGE ： 指定消息提供者在消息接收者没有确认发送时重新发送消息(这种确认模式不在乎接收者收到重复的消息）
        session=conn.createSession(true, Session.AUTO_ACKNOWLEDGE);

        // 创建目标，就创建主题也可以创建队列
       // Destination destination=session.createQueue("queue.hello");//PTP即点对点消息模式
        Destination destination=session.createTopic(ConfigUtil.getConfigByKey("jms.topic"));//采用Sub/Pub即发布/订阅者消息模式
        //消息的消费者
        MessageConsumer consumer=session.createConsumer(destination);
        consumer.setMessageListener(this);

        //等待接收消息
        while(!stop){
            Thread.sleep(5000);
        }

        consumer.close();
        session.close();
        conn.close();
    }

    public void onMessage(Message m) {
        try{
            if(m instanceof TextMessage){//接收文件消息
                TextMessage message=(TextMessage)m;
                System.out.println("收到的消息为："+message.getText());
            }else if(m instanceof MapMessage){//接收键值消息
                MapMessage message=(MapMessage)m;
                System.out.println("收到的消息为：age="+message.getLong("age")+";sarry="+message.getDouble("sarray")+";username="+message.getString("username"));
            }else if(m instanceof StreamMessage){//接收流消息
                StreamMessage message=(StreamMessage)m;
                System.out.println("收到的消息为："+message.readString());
                System.out.println(message.readLong());
            }else if(m instanceof BytesMessage){
                byte[] b=new byte[1024];
                int len=-1;
                BytesMessage message=(BytesMessage)m;
                StringBuilder sb = new StringBuilder();
                while((len=message.readBytes(b))!=-1){
                    sb.append(new String(b,0,len));
                }
                System.out.println("收到的消息为："+sb.toString());
            }else if(m instanceof ObjectMessage){
                System.out.println(m);
            } else{
                System.out.println("收到的消息为："+m);
            }
            session.commit();
        }catch(JMSException e){
            e.printStackTrace();
        }
    }

}
